package org.example.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.Scheduler;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import org.example.config.RegistryConfig;
import org.example.pojo.ServiceMetaInfo;

import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class EtcdRegistry implements Registry{
    private Client client;
    private KV kvClient;

    private static final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * 本机注册的节点 key 集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 本地缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    /**
     * 初始化
     *
     * @param registryConfig
     */
    @Override
    public void init(RegistryConfig registryConfig) {
      client =  Client.builder().endpoints(registryConfig.getAddress()).build();
       kvClient = client.getKVClient();
       heartBeat();
    }

    /**
     * 注册服务
     *
     * @param serviceMetaInfo
     */
    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        // 获取Lease客户端（修正命名：leaseclient → leaseClient，方法名getLeaseclient → getLeaseClient）
        Lease leaseClient = client.getLeaseClient();

        // 创建30秒的租约（修正：grant返回的是CompletableFuture，通过get()阻塞获取结果；getID() → getId()，符合Java方法命名规范）
        long leaseId = leaseClient.grant(30).get().getID();

        // 构建注册键（服务元信息在etcd中的存储路径）
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        // 转换键为ByteSequence（etcd要求的字节序列格式；修正：补充分号，StandardCharsets.UTF 8 → StandardCharsets.UTF_8）
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);

        // 服务元信息序列化为JSON字符串（修正：JSONutil.to]sonStr → JSONUtil.toJsonStr，补充括号和分号）
        String serviceMetaJson = JSONUtil.toJsonStr(serviceMetaInfo);
        ByteSequence value = ByteSequence.from(serviceMetaJson, StandardCharsets.UTF_8);

        // 构建Put选项：关联租约（键随租约过期自动删除）（修正：Putoption → PutOption，putoption → putOption，补充分号）
        PutOption putOption = PutOption.builder()
                .withLeaseId(leaseId) // 绑定租约ID
                .build();

        // 执行put操作：将服务信息存入etcd（阻塞等待结果，确保注册成功）
        kvClient.put(key, value, putOption).get();
        //添加节点信息到本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    /**
     * 注销服务
     *
     * @param serviceMetaInfo
     */
    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registerKey =ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8));
        localRegisterNodeKeySet.remove(registerKey);
    }

    /**
     * 服务发现（获取某服务的所有节点）
     *
     * @param serviceKey
     */
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 从缓存中获取服务列表
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache();
        if (serviceMetaInfos!=null&&!serviceMetaInfos.isEmpty()){
            return serviceMetaInfos;
        }
        // 构建前缀搜索路径
        String searchPrefix = ETCD_ROOT_PATH + serviceKey ;

        try {
            // 构建查询选项：启用前缀搜索
            GetOption getOption = GetOption.builder()
                    .isPrefix(true) // 前缀匹配
                    .build();

            // 执行前缀查询：获取所有匹配前缀的键值对
            // 修正：类名、方法名、变量名大写规范；补全参数括号和分隔符
            GetResponse response = kvClient.get(
                    ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                    getOption
            ).get(); // 阻塞获取结果

            // 获取查询到的键值对列表
            List<KeyValue> keyValues = response.getKvs();

            // 解析键值对为ServiceMetaInfo列表
            List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream()
                    .map(keyValue -> {
                        // 将value从ByteSequence转为JSON字符串（修正toString方法名和字符集）
                        String valueStr = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        // 监听 key 的变化
                        watch(key);
                        // 反序列化为ServiceMetaInfo对象（修正JSONUtil类名拼写）
                        return JSONUtil.toBean(valueStr, ServiceMetaInfo.class);
                    })
                    .collect(Collectors.toList()); // 修正Collectors类名大写
            // 写入缓存
            registryServiceCache.writeCache(serviceMetaInfoList);
                return serviceMetaInfoList;
        } catch (Exception e) {
            // 封装异常信息（修正中文逗号为英文逗号）
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    /**
     * 服务销毁
     */
    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        for (String key : localRegisterNodeKeySet) {
            kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8));
        }
        if (kvClient!=null){
            kvClient.close();
        }
        if (client!=null){
            client.close();
        }
    }
    /**
     * 心跳检测服务
     */
    @Override
    public void heartBeat() {
        // 10 秒续签一次
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                // 遍历本节点所有的 key
                for (String key : localRegisterNodeKeySet) {
                    try {
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
                        // 该节点已过期(需要重启节点才能重新注册)
                        if (CollUtil.isEmpty(keyValues)) {
                            continue;
                        }
                        // 节点未过期，重新注册(相当于续签)
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "续签失败", e);
                    }
                }
            }
        });
        // 支持秒级别定时任务
        CronUtil.setMatchSecond(true);
        Scheduler scheduler = CronUtil.getScheduler();
        if (!scheduler.isStarted()){
            CronUtil.start();
        }

    }

    /**
     * 监听 服务
     *
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        // 之前未被监听，开启监听
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        // key 删除时触发
                        case DELETE:
                            // 清理注册服务缓存
                            registryServiceCache.clearCache();
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
}
